package com.mpgame.common.mq;

import java.util.Properties;
import org.apache.kafka.clients.producer.*;

import com.google.protobuf.Message;
import com.mpgame.common.proto.CmdId;
import com.mpgame.common.proto.ErrorCode;
import com.mpgame.common.proto.MQPacket;
import com.mpgame.common.utils.Log;

public final class KafkaBus {
    private static volatile KafkaProducer<String, MQPacket> msgProducer;
    private static KafkaProducer<String, MQPacket> getMsgProducer(){
        if(msgProducer == null){
            synchronized(KafkaBus.class){
                if(msgProducer == null){
                    Properties props = KafkaCfg.baseProducerProps();
                    msgProducer = new KafkaProducer<>(props);
                }
            }
        }
        return msgProducer;
    }
    public static void sendMsg(int keyInt, int targetId, CmdId msgType, Message payload){
        Log.info("向消息队列产生消息(toplayer)：" + msgType);
        String key = String.valueOf(keyInt);
        MQPacket pkt = MQPacket.newBuilder()
        .addTargetAccountId(targetId)
        .setCmd(msgType)
        .setEc(ErrorCode.SUCCESS)
        .setData(payload.toByteString())
        .build();
        ProducerRecord<String, MQPacket> rec = 
            new ProducerRecord<String,MQPacket>(KafkaTopics.NOTIFY_TO_PLAYER, key, pkt);
        getMsgProducer().send(rec);
    }
    public static void sendServerMsg(int keyInt, CmdId msgType, Message payload){
        Log.info("向消息队列产生消息(toserver)：" + msgType);
        String key = String.valueOf(keyInt);
        MQPacket pkt = MQPacket.newBuilder()
        .setCmd(msgType)
        .setEc(ErrorCode.SUCCESS)
        .setData(payload.toByteString())
        .build();
        ProducerRecord<String, MQPacket> rec = 
            new ProducerRecord<String,MQPacket>(KafkaTopics.NOTIFY_TO_SERVER, key, pkt);
        getMsgProducer().send(rec);
    }
}
